package com.fafian.publish;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PublishBus {
	
	private Map<Class<?>, List<? extends Listener<?>>> listeners;
	private ExecutorService executor;
	private boolean blockNotify;
	protected final Logger logger = LoggerFactory.getLogger(this.getClass());

	
	public PublishBus(boolean blockNotify){
		this.blockNotify = blockNotify;
		listeners = new HashMap<Class<?>, List<? extends Listener<?>>>();
		executor = Executors.newFixedThreadPool(10);
	}
	
	public PublishBus(){
		this(true);
	}
	


	@SuppressWarnings("unchecked")
	public synchronized <T extends Object>void registerListener(Listener<T> toRegister, Class<T> typeOfMessages){
		logger.debug("Adding listener {} for type {}", toRegister, typeOfMessages);
		
		if ((typeOfMessages == null) || (listeners == null) || (toRegister == null)) {
			return;
		}
		
		List<Listener<T>> typeListeners;
		if (!listeners.containsKey(typeOfMessages)){
			logger.debug("First listener for type {}, init ArrayList", typeOfMessages);
			typeListeners = new ArrayList<Listener<T>>();
			listeners.put(typeOfMessages, typeListeners);
		}
		else{
			typeListeners = ((List<Listener<T>>) listeners.get(typeOfMessages));
		}
		typeListeners.add(toRegister);
	}
	
	public synchronized <T>void unRegisterListener(Listener<T> toUnRegister, Class<T> typeOfMessages){
		logger.debug("Removing listener {} for type {}", toUnRegister, typeOfMessages);
		if ((listeners == null) || (toUnRegister == null) || (typeOfMessages == null)){
			return;
		}
	
		if (listeners.containsKey(typeOfMessages)){
			listeners.get(typeOfMessages).remove(toUnRegister);
		}
	}
	
	public synchronized <T>void publish(T value){
		if (listeners == null){
			return;
		}
		if (value==null){
			return;
		}
		Class<?> clazz = value.getClass() ;
		@SuppressWarnings("unchecked")
		List<Listener<T>> clazzListeners = (List<Listener<T>>) listeners.get(clazz);
		if (clazzListeners != null){
			if (!blockNotify){
				for (Listener<T> listener: clazzListeners){
					logger.debug("Submiting a indivitual notify for type {}, value {} and listener {}"
							,clazz, value, listener);
					executor.submit(new IndividualNotifier<T>(listener, value));
				}
			}
			else{
				logger.debug("Submiting a block notify for type {} and value {}",clazz, value);
				executor.submit(new BlockNotifier<T>(clazzListeners, value));
			}
		}
		else
			logger.debug("No listeners for value: {}", value);
	}
	
	private class IndividualNotifier<T> implements Runnable{
		
		private T value;
		private Listener<T> listener;
		
		IndividualNotifier(final Listener<T> listener, final T value ){
			this.value = value;
			this.listener = listener;
		}


		@Override
		public void run() {
			try{
				listener.notify(value);
			}
			catch (Exception e){
				logger.error("Exception {} while notifying listener {}", e, listener);
			}
			
		}
		
	}
	
	private class BlockNotifier<T> implements Runnable{
		private T value;
		private List<Listener<T>> listeners;
		
		BlockNotifier(final  List<Listener<T>> listeners, final T value ){
			this.value = value;
			this.listeners = listeners;
		}


		@Override
		public void run() {
			logger.debug("Start notifying for value: {}", value);
			for (Listener<T> listener: listeners){
				try{
					logger.debug("Notifying: {} - Value: {}", listener, value);
					listener.notify(value);
				}
				catch (Exception e){
					logger.error("Exception {} while notifying listener {}", e, listener);
				}
			}
			logger.debug("Finish notifying for value: {}", value);
		}
		

	}
}
